package com.example.flinktest.transformation;

import com.example.flinktest.source.ClickSource;
import com.example.flinktest.source.Event;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransReduceTest {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //这里ClickSource()使用了之前自定义数据源小节中的ClickSource()
        env.addSource(new ClickSource())
                //将Event数据类型转换成元祖类型
                .map(new MapFunction<Event, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(Event event) throws Exception {
                        return Tuple2.of(event.url, 1L);
                    }
                })
                //使用用户名来进行分流
                .keyBy(r -> r.f0)
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> v1, Tuple2<String, Long> v2) throws Exception {
                        //每到一条数据，用户pv的统计值加1
                        return Tuple2.of(v1.f0, v1.f1 + v2.f1);
                    }
                })
                //为每一条数据分配同一个key,将聚合结果发送到一条流中去
                .keyBy(r -> true)
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> v1, Tuple2<String, Long> v2) throws Exception {
                        //将累加器更新为当前最大的PV统计值，然后向下游发送累加器的值
                        return v1.f1 > v2.f1 ? v1 : v2;
                    }
                }).print();
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
